package com.example.myapplication;

import java.nio.channels.SocketChannel;
import java.util.*;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.lang.Math;
import java.lang.Thread;
import java.lang.reflect.Array;

class myThr extends Thread {

    streamc st;

    public void run() {
        try {

            st.doWork();
        } catch (Exception e) {

            helpFunc.print(e.toString());
        }
    }
}

 class streamc extends streambase {
    Map<DatagramChannel, sockMapData> sockMap = new HashMap<DatagramChannel, sockMapData>();
    Map<Integer, Integer> availPort = new HashMap<Integer, Integer>();
    Map<Integer, Integer> availPort2 = new HashMap<Integer, Integer>();

    LinkedList<Integer> cachePort = new LinkedList<Integer>();
    byte[] salt;
    String ip;
    int statisOut, statisGot;
    Map<Integer, DatagramChannel> reusedPort = new HashMap<Integer, DatagramChannel>();
    double newPortLimit;
    int newPortThisPeriod = 0, newPortThisSecond = 0;
    double maxRecTime;
    LinkedList<Double> maxRecTimeQ = new LinkedList<Double>();
    double minRecTime = Double.MAX_VALUE;
    Map<Double, Integer> newPortMap = new HashMap<Double, Integer>();
    int decreaseDose = 0;
    double startTime;

    double timeoutTime = helpFunc.timeoutTime;
    boolean tooMuchPorts1 = false, tooMuchPorts2 = false;
    logClass myLog;
    int con_portIncreaseDose,speedLimit,perSecondPortsLimit,con_portDecreaseDose,maxPortNum;


    public streamc(String serverip, int increaseDose,int deDose,int pushA,
                   int sizeLi,int limit,int speed,ArrayList<Integer> lp,byte[] salt_a,logClass lo)

            throws Exception {

        super( pushA, sizeLi, false);
        maxPortNum = lp.size();
        myLog = lo;
        salt = salt_a;
        for (int i : lp) {
            if (availPort.size() >= helpFunc.minPortNum) {
                cachePort.add(i);
                continue;
            }
            availPort.put(i, 0);
        }
        ip = serverip;
        statisGot = statisOut = 0;
        newPortLimit = helpFunc.regulateTimeSpan * limit;
        for (int i = 0; i < 20; i++)
            maxRecTimeQ.add((double) 0);
        startTime = helpFunc.getTime();
        speedLimit = speed;
        con_portIncreaseDose = increaseDose;
        perSecondPortsLimit = limit;
        con_portDecreaseDose = deDose;
    }

    int calPara() {
        if (statisGot != 0)
            waitingTime = maxRecTime - minRecTime;
        int dose = 0;
        if (rRaw > speedLimit * 1024) {
            if (statisOut > helpFunc.speedLimit_portNum)
                dose = -1;
        } else
            dose = con_portIncreaseDose;
        newPortLimit = helpFunc.regulateTimeSpan * perSecondPortsLimit;
        if (statisGot < helpFunc.minPackGot)
            sendStatusRate = helpFunc.con_maxRate
                    - (helpFunc.con_maxRate - helpFunc.con_minRate) * ((double) statisGot / helpFunc.minPackGot);
        else
            sendStatusRate = helpFunc.con_minRate;
        if (statisGot == 0)
            timeoutTime = helpFunc.timeoutTime;
        else
            timeoutTime = Collections.max(maxRecTimeQ) + 0.1;
        double lossRate = 1;
        if (statisGot + statisOut != 0)
            lossRate = (double) statisOut / (statisGot + statisOut);

        if (lossRate >= 0 && lossRate <= 0.05) {
            slope = 1;
            maxSend = 3;
        } else if (lossRate >= 0.05 && lossRate <= 0.1) {
            slope = 1.5;
            maxSend = 3;
        } else if (lossRate >= 0.1 && lossRate <= 0.4) {
            slope = 2;
            maxSend = 4;
        } else if (lossRate >= 0.4 && lossRate <= 0.6) {
            slope = 3;
            maxSend = 5;
        } else if (lossRate >= 0.6 && lossRate <= 0.8) {
            slope = 4;
            maxSend = 8;
        } else if (lossRate >= 0.8 && lossRate <= 1) {
            slope = 5;
            maxSend = 10;
        }
        return dose;
    }

    void calNewPortThisPeriod() {
        double t = helpFunc.getTime();
        ArrayList<Double> temp = new ArrayList<Double>(newPortMap.keySet());
        for (double k : temp) {
            if (t - k > helpFunc.regulateTimeSpan)
                newPortMap.remove(k);
        }
        if (newPortMap.size() == 0)
            newPortThisPeriod = 0;
        else {
            newPortThisPeriod = newPortMap.values().stream().mapToInt(Integer::intValue).sum();
        }
    }

    double refreshNewPortTime() {
        double t = helpFunc.getTime();
        if (newPortMap.isEmpty() || newPortMap.values().stream().mapToInt(Integer::intValue).sum() < newPortLimit)
            return t;
        int su = newPortMap.values().stream().mapToInt(Integer::intValue).sum();
        ArrayList<Double> temp = new ArrayList<Double>(newPortMap.keySet());
        Collections.sort(temp);
        for (double i : temp) {
            su -= newPortMap.get(i);
            if (su < newPortLimit)
                return i + helpFunc.regulateTimeSpan;
        }
        return t;
    }

    double refreshNextTimeout() {
        double t = helpFunc.getTime();
        if (sockMap.isEmpty())
            return Double.MAX_VALUE;
        double minT = t;
        for (DatagramChannel k : sockMap.keySet()) {
            sockMapData v = sockMap.get(k);
            if (v.createTime < minT)
                minT = v.createTime;
        }
        return minT + timeoutTime;

    }

    void adjustPortNum(int dose) {
        if (tooMuchPorts1 && !tooMuchPorts2) {
            decreaseDose = con_portDecreaseDose;
            return;
        }
        if (blankRec != 0 && blankSend != 0 && statisGot > helpFunc.minPackGot) {
            decreaseDose = con_portDecreaseDose;
            return;
        }
        if (dose < 0) {
            decreaseDose = -1 * dose;
            return;
        }
        for (int i = 0; i < dose; i++) {
            if (cachePort.size() > 0) {
                int n = cachePort.getFirst();
                cachePort.removeFirst();
                availPort.put(n, 0);
            }
        }
    }

    public void doWork() throws Exception {
        makeOutSo();
        while (true) {
            doWork2();
        }
    }

    public void doWork2() throws Exception {
        if (helpFunc.getTime() - updatedTime > helpFunc.con_close)
            System.exit(0);
        double t = helpFunc.getTime();
        double t1;

        if (availPort.size() != 0) {
            t1 = refreshNewPortTime();
            tooMuchPorts1 = true;
        } else {
            t1 = Double.MAX_VALUE;
            tooMuchPorts2 = true;
        }

        double t2 = refreshNextTimeout();
        double mTime = Math.min(t1, t2);
        double wt;
        if (mTime == Double.MAX_VALUE)
            wt = 1;
        else if (t > mTime) {
            wt = 0;
        } else {
            wt = mTime - t;
        }
        if (sockMap.isEmpty())
            Thread.sleep((long) (wt * 1000));

        else {
            int outSoSign = 0;
            boolean doR = false;
            boolean doW = false;
            if (writeBuffer.length < helpFunc.con_streamBufferSize) {
                outSoSign |= SelectionKey.OP_READ;
                doR = true;
            }
            if (readBuffer.length > 0) {
                outSoSign |= SelectionKey.OP_WRITE;
                doW = true;
            }
            Selector selector = Selector.open();
            if (doR || doW)
                outSo.register(selector, outSoSign);
            for (DatagramChannel channel : sockMap.keySet())
                channel.register(selector, SelectionKey.OP_READ);

            ArrayList<SelectableChannel> rlR = new ArrayList<SelectableChannel>();
            ArrayList<SelectableChannel> wlR = new ArrayList<SelectableChannel>();

            int ret = selector.select(1000);
            Iterator selectedKeys = selector.selectedKeys().iterator();
            while (selectedKeys.hasNext()) {
                SelectionKey key = (SelectionKey) selectedKeys.next();
                selectedKeys.remove();
                if (!key.isValid()) {
                    continue;
                }
                if (key.channel() == outSo && key.isReadable() && doR)
                    rlR.add(key.channel());
                if (key.channel() == outSo && key.isWritable() && doW)
                    wlR.add(key.channel());
                if (key.channel() != outSo && key.isReadable())
                    rlR.add(key.channel());
            }
            if (!wlR.isEmpty()) {
                ByteBuffer temp = ByteBuffer.wrap(readBuffer);
                outSo.write(temp);
                readBuffer = new byte[temp.remaining()];
                temp.get(readBuffer, 0, readBuffer.length);

            }
            if (rlR.contains(outSo)) {
                ByteBuffer buf = ByteBuffer.allocate(helpFunc.con_streamBufferSize - writeBuffer.length);

                int bytesRead = outSo.read(buf);
                byte[] temp = new byte[writeBuffer.length + bytesRead];
                System.arraycopy(writeBuffer, 0, temp, 0, writeBuffer.length);
                System.arraycopy(buf.array(), 0, temp, writeBuffer.length, bytesRead);
                writeBuffer = temp;
            }
            ArrayList<byte[]> re = deal_rec(rlR);
            deal_data_back(re);
            selector.close();
        }
        read();
        write();
        deal_timeout();
        calNewPortThisPeriod();
        int l = availPort.size();

        if (l + newPortThisPeriod > newPortLimit) {
            l = (int) newPortLimit - newPortThisPeriod;
            if (l < 0)
                l = 0;
        }
        int sendNum = l + availPort2.size();

        ArrayList<byte[]> re = get_data_to_send(sendNum);
        sendData(re);
        if (helpFunc.getTime() - staTime > 1) {
            staTime = helpFunc.getTime();

            String log = String.format("%3d %4d %4d|%4d %4d <s,r>%5d %5d %5d %5d",
                    sockMap.size(),	statisGot,statisOut,(int)(rRaw/1024),(int)(wRaw/1024),
                    helpFunc.getPackStaBigV(maxSendL),helpFunc.getPackStaBigV(peerMaxRec),
                    helpFunc.getPackStaBigV(peerMaxSend),helpFunc.getPackStaBigV(maxRec));



            myLog.setLog(log);

            newPortThisSecond = 0;
            maxRecTimeQ.add(maxRecTime);
            maxRecTimeQ.removeFirst();

            int dose = calPara();
            adjustPortNum(dose);

            tooMuchPorts1 = tooMuchPorts2 = false;
            statisGot = statisOut = 0;
            maxRecTime = 0;
            minRecTime = Double.MAX_VALUE;
            rRaw = wRaw = rNet = wNet = 0;
            totalRec = blankRec = totalSend = blankSend = statusRev = statusSend = 0;
            helpFunc.clearPackSta(maxSendL);
            helpFunc.clearPackSta(maxRec);
            helpFunc.clearPackSta(peerMaxRec);
            helpFunc.clearPackSta(peerMaxSend);
        }
    }
    boolean retireSock(int n) {
        if (cachePort.size() >= maxPortNum - helpFunc.minPortNum || decreaseDose == 0)
            return false;
        decreaseDose -= 1;
        if (availPort.size() > 0) {
            ArrayList<Integer> n2 = new ArrayList<Integer>(availPort.keySet());
            availPort.remove(n2.get(0));
            cachePort.add(n2.get(0));
            return false;
        }
        cachePort.add(n);
        return true;
    }
    void sendData(ArrayList<byte[]> re) throws Exception {
        int co = -1;
        int l = re.size();
        double ft = helpFunc.getTime();
        ArrayList<Integer> tt = new ArrayList<Integer>(availPort2.keySet());
        tt.addAll(availPort.keySet());
        for (int k : tt) {
            co += 1;
            if (co == l)
                return;
            byte[] data = re.get(co);
            DatagramChannel sock;
            if (reusedPort.containsKey(k)) {
                sock = reusedPort.get(k);
                reusedPort.remove(k);
            } else {
                sock = DatagramChannel.open();
                sock.configureBlocking(false);
                newPortThisSecond += 1;
                if (newPortMap.containsKey(ft))
                    newPortMap.put(ft, newPortMap.get(ft) + 1);
                else
                    newPortMap.put(ft, 1);
            }
            sockMapData dd = new sockMapData();
            dd.num = k;
            dd.createTime = helpFunc.getTime();
            sockMap.put(sock, dd);
            helpFunc.makePack(data, salt);
            sockMap.get(sock).uuid = helpFunc.uuid;
            InetSocketAddress add = new InetSocketAddress(ip, k);
            ByteBuffer bb = ByteBuffer.wrap(helpFunc.con);
            sock.send(bb, add);
            helpFunc.addPackSta(maxSendL, (short) helpFunc.con.length, 0);
            wRaw += helpFunc.con.length;
            if (availPort.containsKey(k))
                availPort.remove(k);
            else
                availPort2.remove(k);
        }
    }
    void deal_timeout() throws Exception {
        ArrayList<DatagramChannel> temp = new ArrayList<DatagramChannel>(sockMap.keySet());
        for (DatagramChannel sock : temp) {
            sockMapData v = sockMap.get(sock);
            if (v.createTime + timeoutTime < helpFunc.getTime()) {
                sock.close();
                int n = sockMap.get(sock).num;
                sockMap.remove(sock);
                if (!retireSock(n))
                    availPort.put(n, 0);
                statisOut += 1;
            }
        }
    }


    ArrayList<byte[]> deal_rec(ArrayList<SelectableChannel> l) throws Exception {
        ArrayList<byte[]> re = new ArrayList<byte[]>();
        for (SelectableChannel sock : l) {
            if (sock == outSo)
                continue;
            DatagramChannel sock2 = (DatagramChannel) sock;
            ByteBuffer dd = ByteBuffer.allocate(helpFunc.recLen);
            sock2.receive(dd);

            byte[] j = new byte[dd.position()];
            System.arraycopy(dd.array(), 0, j, 0, dd.position());
            rRaw += j.length;
            helpFunc.checkPackValid2(j, salt);
            byte[] ub = sockMap.get(sock2).uuid;
            int n = sockMap.get(sock2).num;
            double ti = sockMap.get(sock2).createTime;
            if (!Arrays.equals(helpFunc.uuid, ub)) {
                sock2.close();
                sockMap.remove(sock2);
                availPort.put(n, 0);
                statisOut += 1;
                continue;
            } else {
                helpFunc.addPackSta(maxRec, (short) j.length, 0);
                double recT = helpFunc.getTime() - ti;
                if (recT > maxRecTime) {
                    maxRecTime = recT;
                    if (maxRecTime > timeoutTime)
                        maxRecTime = timeoutTime;
                }
                if (recT < minRecTime)
                    minRecTime = recT;
                if (retireSock(n))
                    sock2.close();
                else {
                    availPort2.put(n, 0);
                    reusedPort.put(n, sock2);
                }
                sockMap.remove(sock2);
                statisGot += 1;

                re.add(helpFunc.con);
            }
        }
        // helpFunc.print("got num"+Integer.toString(statisGot));
        return re;
    }


    public static void main(String[] args) throws Exception
    {
/*
		SocketChannel so = SocketChannel.open();
		ArrayList<Integer> listenPort = new ArrayList<Integer>();
		for (int i = 50000; i < 50100; i++)
			listenPort.add(i);
		String ss = "salt";
		streamc st = new streamc(listenPort, ss.getBytes(), helpFunc.con_minRate, helpFunc.pushAhead,
				helpFunc.packLimit);
		myThr aa = new myThr();
		aa.st = st;
		aa.setDaemon(true);
		aa.start();
		InetSocketAddress add = new InetSocketAddress("localhost", st.listenPort);
		so.connect(add);
		System.out.println("ok");
		while (true) {
			byte [] sss = new byte[1000];
			ByteBuffer hehe = ByteBuffer.wrap(sss);
			so.write(hehe);
		}
*/
    }


}
